#include "my_disk_volume.h"

#define MY_DISK_VOLUME_SHM_PREFIX           "my_disk_volume_shm_"

// TODO: to be supported volume timeout detection
// 写超时探测: 在buf flush操作中绑定一个定时器, 用于检测写操作是否超时.
// 读操作探测: volume定期向aio_manager发起一个空任务, 检测aio_manager是否长时间阻塞.
// 并发度监控: volume定期检测活跃的事务数, 超过一定限制时上报异常. (避免任务堆积)

static void
my_disk_volume_source_release(my_disk_volume_t *volume) {
    // my_disk_data_close needs to ensure reentry security
    my_disk_data_close(&volume->disk_data);

    // my_disk_index_close needs to ensure reentry security
    my_disk_index_close(&volume->index);

    if (volume->hotspot) {
        my_disk_hotspot_close(volume->hotspot);
        my_free(volume->hotspot);
        volume->hotspot = NULL;
    }

    // my_aio_manager_wait_exit needs to ensure reentry security
    if (volume->aio_manager) {
        my_aio_manager_wait_exit(volume->aio_manager);
        my_free(volume->aio_manager);
        volume->aio_manager = NULL;
    }

    if (volume->tmp_buf) {
        my_free(volume->tmp_buf);
        volume->tmp_buf = NULL;
    }

    if (volume->shm_addr) {
        my_syscall_munmap(volume->shm_addr, volume->shm_size);
        volume->shm_addr = NULL;
    }

    if (volume->shm_fd != INVALID_FD) {
        my_syscall_close(volume->shm_fd);
        volume->shm_fd = INVALID_FD;
    }

    if (volume->shm_name) {
        my_free(volume->shm_name);
        volume->shm_name = NULL;
    }

    if (volume->name) {
        my_free(volume->name);
        volume->name = NULL;
    }
}

static void
my_disk_volume_open_done(my_disk_volume_t *volume, my_bool_t success) {
    if (!success) {
        MY_LOG_DEBUG0("disk_volume", "open disk volume failed, will release volume source");
        my_disk_volume_source_release(volume);
        return volume->done_cb(volume, MY_FALSE);
    }

    my_cache_config_hotspot_t *hotspot = &g_cache_conf.storage_cache.hotspot;

    if (hotspot->enabled && hotspot->hot_ratio > 0) {
        volume->hotspot = my_malloc(sizeof(my_disk_hotspot_t));
        int64_t hotspot_max_items = volume->item_num * hotspot->hot_ratio;
        int64_t hotspot_max_bytes = volume->data_size * hotspot->hot_ratio;
        my_disk_hotspot_open(volume->hotspot, hotspot->nlist,
            hotspot_max_items, hotspot_max_bytes);
    }

    volume->vinfo.fd = volume->fd;
    volume->vinfo.header_offset = volume->start_offset;
    volume->vinfo.index_offset = volume->index_offset;
    volume->vinfo.data_offset = volume->data_offset;
    volume->vinfo.data_size = volume->data_size;
    my_disk_volume_header_t *header = (my_disk_volume_header_t *)volume->shm_addr;
    volume->vinfo.write_serial = header->write_serial;
    volume->vinfo.recycle_serial = header->recycle_serial;
    volume->vinfo.write_pos = header->write_pos;
    volume->vinfo.recycle_pos = header->recycle_pos;
    volume->vinfo.volume_header = header;

    my_disk_index_init(&volume->index);
    if (!my_disk_index_open(&volume->index, &volume->vinfo,
        volume->shm_addr + volume->header_size, volume->index_size)) {
        my_disk_volume_source_release(volume);
        return volume->done_cb(volume, MY_FALSE);
    }

    if (!my_disk_data_open(&volume->disk_data, &volume->vinfo, volume->hotspot,
        volume->aio_manager, volume, my_disk_volume_reinsert)) {
        my_disk_volume_source_release(volume);
        return volume->done_cb(volume, MY_FALSE);
    }

    return volume->done_cb(volume, MY_TRUE);
}

static void
my_disk_volume_write_header_and_index_done(my_dispatch_event_t *dispatch_event) {
    my_aio_req_t *aio_req = dispatch_event->data;
    my_disk_volume_t *volume = dispatch_event->save_data;
    if (aio_req->eno != 0 || aio_req->nbytes != aio_req->res) {
        MY_LOG_WARN("disk_volume", "write volume header and index failed, errno: %d",
            aio_req->eno);
        return my_disk_volume_open_done(volume, MY_FALSE);
    }

    return my_disk_volume_open_done(volume, MY_TRUE);
}

static void
my_disk_volume_create_new(my_disk_volume_t *volume) {
    my_disk_volume_header_t header = {
        .tag = MY_DISK_VOLUME_HEADER_TAG,
        .version = MY_DISK_VOLUME_LAYOUT_VERSION,
        .block_size = MY_DISK_BLOCK_SIZE,
        .index_item_num = volume->item_num,
        .write_serial = 0,
        .recycle_serial = 0,
        .write_pos = 0,
        .recycle_pos = 0
    };
    my_disk_volume_header_update_checksum(&header);

    // clear the index in shm
    bzero(volume->shm_addr + sizeof(my_disk_volume_header_t),
        volume->shm_size - sizeof(my_disk_volume_header_t));
    
    // reset volume header
    *(my_disk_volume_header_t *)volume->shm_addr = header;

    // write volume header and index table
    my_aio_req_init(&volume->aio_req, AIO_WRITE, volume->fd, volume->shm_addr,
        volume->shm_size, volume->start_offset);
    volume->dispatch_event.handler = my_disk_volume_write_header_and_index_done;
    volume->dispatch_event.data = &volume->aio_req;
    volume->dispatch_event.save_data = volume;
    my_aio_manager_post_event(volume->aio_manager, &volume->dispatch_event);
}

static void
my_disk_volume_read_index_done(my_dispatch_event_t *dispatch_event) {
    my_aio_req_t *aio_req = dispatch_event->data;
    my_disk_volume_t *volume = dispatch_event->save_data;
    if (aio_req->eno != 0 || aio_req->nbytes != aio_req->res) {
        MY_LOG_WARN("disk_volume", "read index failed, errno: %d", aio_req->eno);
        return my_disk_volume_open_done(volume, MY_FALSE);
    }

    // set volume header
    *(my_disk_volume_header_t *)volume->shm_addr =
        *(my_disk_volume_header_t *)volume->tmp_buf;

    return my_disk_volume_open_done(volume, MY_TRUE);
}

static void
my_disk_volume_read_index(my_disk_volume_t *volume) {
    char *shm_index_buf = volume->shm_addr + volume->header_size;
    my_aio_req_init(&volume->aio_req, AIO_READ, volume->fd, shm_index_buf,
        volume->index_size, volume->index_offset);
    volume->dispatch_event.handler = my_disk_volume_read_index_done;
    volume->dispatch_event.data = &volume->aio_req;
    volume->dispatch_event.save_data = volume;
    my_aio_manager_post_event(volume->aio_manager, &volume->dispatch_event);
}

static void
my_disk_volume_read_header_done(my_dispatch_event_t *dispatch_event) {
    my_aio_req_t *aio_req = dispatch_event->data;
    my_disk_volume_t *volume = dispatch_event->save_data;
    if (aio_req->eno != 0 || aio_req->nbytes != aio_req->res) {
        MY_LOG_WARN("disk_volume", "read disk volume header failed, errno: %d", aio_req->eno);
        return my_disk_volume_open_done(volume, MY_FALSE);
    }

    my_disk_volume_header_t *header = (my_disk_volume_header_t *)volume->tmp_buf;
    if (my_disk_volume_header_check(header, volume->item_num)) {
        return my_disk_volume_read_index(volume);
    }

    return my_disk_volume_create_new(volume);
}

static void
my_disk_volume_check_and_fix_header(my_disk_volume_t *volume) {
    my_disk_volume_header_t *header = (my_disk_volume_header_t *)volume->shm_addr;
    if (my_disk_volume_header_check(header, volume->item_num)) {
        return my_disk_volume_open_done(volume, MY_TRUE);
    }

    // read volume header in disk file
    volume->tmp_buf = my_malloc(volume->header_size);
    my_aio_req_init(&volume->aio_req, AIO_READ, volume->fd, volume->tmp_buf,
        volume->header_size, volume->start_offset);
    volume->dispatch_event.handler = my_disk_volume_read_header_done;
    volume->dispatch_event.data = &volume->aio_req;
    volume->dispatch_event.save_data = volume;

    my_aio_manager_post_event(volume->aio_manager, &volume->dispatch_event);
}

static my_bool_t
my_disk_volume_open_shm_helper(my_disk_volume_t *volume) {
    my_syscall_result_t res = my_syscall_shm_open(volume->shm_name, O_RDWR | O_CREAT, 0640);
    if (res.eno != 0) {
        MY_LOG_WARN("disk_volume", "open %s shm failed, errno: %d", volume->shm_name, res.eno);
        return MY_FALSE;
    }
    volume->shm_fd = (int)res.res;

    my_file_stat_t stat_buf;
    res = my_syscall_file_fstat(volume->shm_fd, &stat_buf);
    if (res.eno != 0) {
        MY_LOG_WARN("disk_volume", "get %s shm file size failed, errno: %d", volume->shm_name, res.eno);
        return MY_FALSE;
    }

    int64_t file_size = stat_buf.st_size;
    if (file_size == volume->shm_size) {
        MY_LOG_DEBUG("disk_volume", "open %s shm file success", volume->shm_name);
        return MY_TRUE;
    }

    res = my_syscall_ftruncate(volume->shm_fd, volume->shm_size);
    if (res.eno != 0) {
        MY_LOG_WARN("disk_volume", "ftruncate %s shm file failed, errno: %d", volume->shm_name, res.eno);
        return MY_FALSE;
    }

    return MY_TRUE;
}

static my_bool_t
my_disk_volume_open_shm(my_disk_volume_t *volume) {
    if (my_disk_volume_open_shm_helper(volume)) {
        return MY_TRUE;
    }

    if (volume->shm_fd != INVALID_FD) {
        my_syscall_close(volume->shm_fd);
        volume->shm_fd = INVALID_FD;
    }

    my_syscall_shm_unlink(volume->shm_name);

    return my_disk_volume_open_shm_helper(volume);
}

void
my_disk_volume_open(my_disk_volume_t *volume, const char *disk_name,
    int volume_id, int fd, int64_t start_offset, int64_t volume_size, my_bool_t create_new) {

    int64_t avg_file_size = g_cache_conf.storage_cache.avg_file_size;
    int64_t item_num = volume_size / avg_file_size;
    volume->disk_name = disk_name;
    volume->fd = fd;
    volume->id = volume_id;
    volume->start_offset = start_offset;
    volume->volume_size = volume_size;
    volume->item_num = item_num;
    volume->header_size = sizeof(my_disk_volume_header_t);
    volume->index_size = my_disk_volume_index_table_size(item_num);
    volume->shm_size = volume->header_size + volume->index_size;
    volume->index_offset = start_offset + volume->header_size;
    volume->data_offset = volume->index_offset + volume->index_size;
    volume->data_size = volume->volume_size - volume->header_size - volume->index_size;
    volume->data_size = (volume->data_size >> MY_PAGE_BITS) << MY_PAGE_BITS; // aligned

    // volume_name  = disk_name + "_" + volume_id
    int volume_name_size = strlen(disk_name) + 128;\
    volume->name = my_malloc(volume_name_size);
    snprintf(volume->name, volume_name_size, "%s-%d", disk_name, volume_id);

    // shm_name = shm_prefix + volume_name
    int shm_name_size = strlen(volume->name) + 128;
    volume->shm_name = my_malloc(shm_name_size);
    snprintf(volume->shm_name, shm_name_size, "%s%s", MY_DISK_VOLUME_SHM_PREFIX,
        volume->name);

    if (!my_disk_volume_open_shm(volume)) {
        MY_LOG_WARN0("disk_volume", "open disk volume shm file failed");
        return my_disk_volume_open_done(volume, MY_FALSE);
    }

    my_syscall_result_t res = my_syscall_mmap(NULL, volume->shm_size, PROT_READ|PROT_WRITE,
        MAP_SHARED, volume->shm_fd, 0);
    if (res.eno != 0) {
        MY_LOG_WARN("disk_volume", "mmap shm %s failed, errno: %d", volume->shm_name, res.eno);
        return my_disk_volume_open_done(volume, MY_FALSE);
    }
    volume->shm_addr = (void *)(intptr_t)res.res;

    volume->aio_manager = my_malloc(sizeof(my_aio_manager_t));
    my_aio_manager_init(volume->aio_manager);
    int aio_threads = g_cache_conf.storage_cache.aio_threads;
    my_aio_manager_start(volume->aio_manager, aio_threads);

    if (!create_new) {
        return my_disk_volume_check_and_fix_header(volume);
    }

    return my_disk_volume_create_new(volume);
}

static void
my_disk_volume_put_done(my_disk_write_context_t *context) {
    my_cache_req_t *req = context->req;
    my_disk_volume_t *volume = context->ctx_data;

    if (req->err = CACHE_ERR_IO_ERR && volume->err_cb) {
        volume->err_cb(volume);
    }

    if (req->err == CACHE_ERR_NO_ERR) {
        my_disk_index_item_t item;
        my_disk_index_item_init(&item, req->key, context->offset,
            context->aligned_size, context->write_serial);
        my_disk_index_insert(&volume->index, req->key, &item);
    }

    // there is lock_node, which means it has been locked.
    // we should unlock it and free it.
    if (context->lock_node) {
        my_disk_key_lock_unlock(context->lock_node);
        my_free(context->lock_node);
    }

    my_free(context);
    return req->handler(req);
}

static void
my_disk_volume_put_lock_done(my_disk_key_lock_node_t *key_lock_node) {
    my_disk_write_context_t *context = key_lock_node->ctx_data;
    my_disk_volume_t *volume = context->ctx_data;
    my_disk_data_write(&volume->disk_data, context);
}

void
my_disk_volume_put(my_disk_volume_t *volume, my_cache_req_t *req) {
    my_disk_write_context_t *context = my_malloc(sizeof(my_disk_write_context_t));
    my_disk_write_context_init(context);
    context->req = req;
    context->ctx_data = volume;
    context->handler = my_disk_volume_put_done;

    if (my_disk_vinfo_closed_write(&volume->vinfo)) {
        req->err = CACHE_ERR_CLOSED;
        req->errmsg = "cache already closed";
        return my_disk_volume_put_done(context);
    }

    my_disk_key_lock_node_t *lock_node = my_malloc(sizeof(my_disk_key_lock_node_t));
    my_disk_key_lock_node_init(lock_node, &volume->lock_tab, req,
        my_disk_volume_put_lock_done, context);
    context->lock_node = lock_node;

    my_disk_key_lock_lock(lock_node);
}

void // the reinsert operation is an additional check on the put operation
my_disk_volume_reinsert_lock_done(my_disk_key_lock_node_t *key_lock_node) {
    my_disk_write_context_t *context = key_lock_node->ctx_data;
    my_disk_volume_t *volume = context->ctx_data;
    my_cache_req_t *req = context->req;

    my_disk_index_item_t item;
    bzero(&item, sizeof(my_disk_index_item_t));
    if (!my_disk_index_lookup(&volume->index, req->key, &item)) {
        req->err = CACHE_ERR_NOT_FOUND;
        req->errmsg = "not found this cache, end of reinsert task";
        return my_disk_volume_put_done(context);
    }

    int64_t offset = my_disk_index_item_offset(&item);
    if (offset != context->old_offset) {
        req->err = CACHE_ERR_ALREADY_UPDATE;
        req->errmsg = "cache object maybe updated, end of reinsert task";
        return my_disk_volume_put_done(context);
    }

    my_disk_data_write(&volume->disk_data, context);
}

void
my_disk_volume_reinsert(my_disk_volume_t *volume, my_cache_req_t *req,
    int64_t old_offset) {

    my_disk_write_context_t *context = my_malloc(sizeof(my_disk_write_context_t));
    my_disk_write_context_init(context);
    context->req = req;
    context->ctx_data = volume;
    context->handler = my_disk_volume_put_done; // code reuse
    context->old_offset = old_offset;

    if (my_disk_vinfo_closed_write(&volume->vinfo)) {
        req->err = CACHE_ERR_CLOSED;
        req->errmsg = "cache already closed";
        return my_disk_volume_put_done(context);
    }

    my_disk_key_lock_node_t *lock_node = my_malloc(sizeof(my_disk_key_lock_node_t));
    my_disk_key_lock_node_init(lock_node, &volume->lock_tab, req,
        my_disk_volume_reinsert_lock_done, context);
    context->lock_node = lock_node;

    my_disk_key_lock_lock(lock_node);
}

static void
my_disk_volume_get_done(my_disk_read_context_t *context) {
    my_cache_req_t *req = context->req;
    my_disk_volume_t *volume = context->ctx_data;

    if (req->err = CACHE_ERR_IO_ERR && volume->err_cb) {
        volume->err_cb(volume);
    }

    if (req->err != CACHE_ERR_NO_ERR &&
        req->err != CACHE_ERR_NOT_FOUND &&
        req->err != CACHE_ERR_CLOSED) {
        my_disk_index_remove(&volume->index, req->key);
    }

    // there is lock_node, which means it has been locked.
    // we should unlock it and free it.
    if (context->lock_node) {
        my_disk_key_lock_unlock(context->lock_node);
        my_free(context->lock_node);
    }

    my_free(context);
    return req->handler(req);
}

static void
my_disk_volume_get_lock_done(my_disk_key_lock_node_t *key_lock_node) {
    my_disk_read_context_t *context = key_lock_node->ctx_data;
    my_disk_volume_t *volume = context->ctx_data;
    my_cache_req_t *req = context->req;

    my_disk_index_item_t item;
    bzero(&item, sizeof(my_disk_index_item_t));
    if (!my_disk_index_lookup(&volume->index, req->key, &item)) {
        req->err = CACHE_ERR_NOT_FOUND;
        req->errmsg = "not found this cache";
        return my_disk_volume_get_done(context);
    }

    context->write_serial = item.serial;
    context->offset = my_disk_index_item_offset(&item);
    context->aligned_size = my_disk_index_item_size(&item);
    return my_disk_data_read(&volume->disk_data, context);
}

// TODO: 应该保证get操作的内容不会被flush操作所覆盖. 这需要引入额外的同步措施.
// 最坏情况下降导致缓存数据无效, crc32可以发现该问题并删除对应的key.
// 整体来说该问题的影响程度较小, 暂不修复.
void
my_disk_volume_get(my_disk_volume_t *volume, my_cache_req_t *req) {
    my_disk_read_context_t *context = my_malloc(sizeof(my_disk_read_context_t));
    my_disk_read_context_init(context);
    context->req = req;
    context->ctx_data = volume;
    context->handler = my_disk_volume_get_done;

    if (my_disk_vinfo_closed(&volume->vinfo)) {
        req->err = CACHE_ERR_CLOSED;
        req->errmsg = "cache already closed";
        return my_disk_volume_get_done(context);
    }

    my_disk_key_lock_node_t *lock_node = my_malloc(sizeof(my_disk_key_lock_node_t));
    my_disk_key_lock_node_init(lock_node, &volume->lock_tab, req,
        my_disk_volume_get_lock_done, context);
    context->lock_node = lock_node;

    my_disk_key_lock_lock(lock_node);
}